package com.study;

import org.apache.flink.api.common.serialization.SerializationSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.connectors.rabbitmq.RMQSink;
import org.apache.flink.streaming.connectors.rabbitmq.common.RMQConnectionConfig;

/**
 * @author sjw
 * @description
 * @date 2019-08-19 下午3:19
 */
public class SourceDefine {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStream<String> dataStream = env.addSource(new SourceFunction<String>() {
            private static final long serialVersionUID = 1L;

            private volatile boolean isRunning = true;

            @Override
            public void run(SourceContext<String> sourceContext) throws Exception {
                while (isRunning) {
                    sourceContext.collect(String.valueOf(Math.floor(Math.random() * 100)));
                }
            }

            @Override
            public void cancel() {
                isRunning = false;
            }

        });
        RichSourceFunction a;
        final RMQConnectionConfig connectionConfig = new RMQConnectionConfig
                .Builder().setHost("localhost").setVirtualHost("/")
                .setPort(5672).setUserName("root").setPassword("123456")
                .build();

        dataStream.addSink(new RMQSink<>(connectionConfig,"q.study.flink.sink", new SerializationSchema() {
            @Override
            public byte[] serialize(Object o) {
                return o.toString().getBytes();
            }
        }));
        env.execute("flink learning connectors rabbitmq");
    }
}
